package com.wifi;

import org.apache.flume.Context;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
import org.apache.flume.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.gson.Gson;
import javax.crypto.Cipher;
import javax.crypto.spec.IvParameterSpec;
import javax.crypto.spec.SecretKeySpec;
import java.io.BufferedReader;
import java.io.DataOutputStream;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
import java.security.KeyFactory;
import java.security.PublicKey;
import java.security.spec.X509EncodedKeySpec;
import java.util.*;

import static com.wifi.ConstantUtils.AES_KEY;
import static com.wifi.ConstantUtils.BAIDU_DOMAIN;
import static com.wifi.ConstantUtils.FORMAL_URL;

/**
 * @Auther: dld
 * @Date: 13:46 2017/11/27
 * @DES:
 * @Modified By:
 */
public class BaiduApiSink extends AbstractSink implements Configurable {

    private static Logger log = LoggerFactory.getLogger(BaiduApiSink.class);
    private static byte[] rsaPublicKey = getRsaPublicKey();

    @Override
    public void configure(Context context) {

    }

    @Override
    public Status process() throws EventDeliveryException {
        Status status = Status.READY;
        Transaction tx = null;
        try {
            List<BaiduDomainVo> list = new ArrayList<BaiduDomainVo>(1);
            BaiduWifiVo baiduWifiVo = new BaiduWifiVo();
            Channel channel = getChannel();
            tx = channel.getTransaction();
            tx.begin();

            for (int i = 0; i < 100; i++) {
                Event event = channel.take();
                if (event == null) {
                    status = Status.BACKOFF;
                    break;
                } else {
                    /*
                    华为数据结构说明
接收到数据时系统时间戳,供应商id,版本,采集设备mac地址(WiFi mac地址),一次上报的探测到的设备的个数,侦测到的设备的mac,侦测到的设备的ip,
设备所在信道,radio类型0->2.4G 1->5G和是否关联本ap 0->否 1->是(打印的值是16进制的原值,目前这个字段确定不了),场强,底噪,
                     */
                    String body = new String(event.getBody());
                    String[] arrs = body.split(",");
                    //1513479396,HW,3,14:a0:f8:da:17:00,7,00:03:7f:00:00:00,0.0.0.0,1,0,-68,-92,64
                    String usr_cap_time = arrs[0];//usr_cap_time: 抓取用户时间的unix时间戳，单位是秒
                    String dev_mac = arrs[3];//dev_mac: 抓取用户的AP设备MAC，需要绑定到具体门店，在绑定门店功能中创建并绑定门店
                    String usr_mac = arrs[5];//usr_mac: 抓取到的用户mac，AP抓到的所有mac地址不一定都是合法的用户mac，假mac会被过滤
                    String usr_distance = arrs[arrs.length - 2];//usr_distance: 抓取时信号强度的相反数
                    // longitude: AP设备的地理经度坐标
                    // latitude: AP设备的地理纬度坐标
                    // dev_ssid: AP设备接入的网络名称

                    BaiduDomainVo baiduDomainVo = new BaiduDomainVo();
                    baiduDomainVo.setDomain(BAIDU_DOMAIN);
                    baiduDomainVo.setDev_mac(dev_mac);
                    baiduDomainVo.setUsr_mac(usr_mac);
                    baiduDomainVo.setUsr_cap_time(Long.parseLong(usr_cap_time));
                    baiduDomainVo.setUsr_distance(- Integer.parseInt(usr_distance));//usr_distance: 抓取时信号强度的相反数
                    list.add(baiduDomainVo);
//                    log.info("第" + i + "次发送数据,发送内容->{}", baiduWifiVo.toString());
                }
            }

            if(list.size() > 0){
                baiduWifiVo.setPayload(list);
                sendData(baiduWifiVo);
                log.info("rows.size->{}",list.size());
            }
            tx.commit();
        } catch (Exception e) {
            if (tx != null) {
                tx.commit();
            }
            e.printStackTrace();
        } finally {
            if (tx != null) {
                tx.close();
            }
        }
        return status;
    }

    public static void main(String[] args) {
        List<BaiduDomainVo> list = new ArrayList<>(1);
        BaiduWifiVo baiduWifiVo = new BaiduWifiVo();
//      log.info("rows.size->{}",rows.size());
        for(int i = 0; i < 100; i++) {
            BaiduDomainVo baiduDomainVo = new BaiduDomainVo();
            baiduDomainVo.setDomain(BAIDU_DOMAIN);
            baiduDomainVo.setDev_mac("9C:50:EE:C6:21:E0");
            baiduDomainVo.setUsr_mac("91:11:D1:01:91:31");//sniffed_mac
            baiduDomainVo.setUsr_cap_time(System.currentTimeMillis() / 1000);
            list.add(baiduDomainVo);
            baiduWifiVo.setPayload(list);
            sendData(baiduWifiVo);
            log.info("第" + i + "次发送数据,发送内容->{}", baiduWifiVo.toString());
            list.clear();
            try {
                Thread.sleep(3 * 1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void sendData(BaiduWifiVo baiduWifiVo){
//        log.info("百度实时传送数据baiduWifiVo:{}",baiduWifiVo);
        String aesKey = AES_KEY;
        byte[] aesKeyEnc = rsaEncrypt(aesKey.getBytes(), rsaPublicKey);
        if (aesKeyEnc == null) {
            return;
        }
        Gson gson = new Gson();
        String json = gson.toJson(baiduWifiVo);
//        if(log.isInfoEnabled()){
//            log.info("baiduWifiVo->{}",json);
//        }
        byte[] jsonEnc = aesEncrypt(json.getBytes(), aesKey.getBytes());
        if (jsonEnc == null) {
            return;
        }

        int code = postData(aesKeyEnc, jsonEnc);
//        if(log.isInfoEnabled()){
//            log.info("Return Code = " + code);
//        }
    }


    private static int postData(byte[] aesKeyEnc, byte[] jsonEnc) {
        try {
            String url = FORMAL_URL;
            HttpURLConnection httpURLConnection = (HttpURLConnection) new URL(url).openConnection();
            httpURLConnection.setDoOutput(true);
            httpURLConnection.setDoInput(true);
            httpURLConnection.setUseCaches(false);
            httpURLConnection.setRequestMethod("POST");
            httpURLConnection.setRequestProperty("Content-Type", "text/plain; charset=UTF-8");

            DataOutputStream dos = new DataOutputStream(httpURLConnection.getOutputStream());
            dos.writeInt(0x484D3031);
            dos.writeInt(1);
            dos.writeInt(1017);
            dos.writeLong(0);
            dos.writeShort(2);
            dos.writeShort(1);
            dos.writeInt(0x484D3031);
            dos.writeInt(aesKeyEnc.length);
            dos.write(aesKeyEnc);
            dos.writeInt(jsonEnc.length);
            dos.write(jsonEnc);
            dos.flush();
            dos.close();
            return httpURLConnection.getResponseCode();
        } catch (Exception e) {
            e.printStackTrace();
            return -1;
        }
    }

    private static byte[] aesEncrypt(byte[] input, byte[] aesKey) {
        try {
            byte[] aesIv = {0x01, 0x01, 0x01, 0x1, 0x01, 0x01, 0x01, 0x1, 0x01, 0x01, 0x01, 0x1, 0x01, 0x01, 0x01, 0x1};
            Cipher cipher = Cipher.getInstance("AES/CBC/PKCS5Padding");
            cipher.init(Cipher.ENCRYPT_MODE, new SecretKeySpec(aesKey, "AES"), new IvParameterSpec(aesIv));
            return cipher.doFinal(input);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    private static byte[] getRsaPublicKey() {
        try {
            String key = "";

            BufferedReader br = new BufferedReader(new InputStreamReader(OpenClientDemo.class.getClassLoader().getResourceAsStream("publickey")));
            String line;
            while ((line = br.readLine()) != null) {
                if (!line.startsWith("-")) {
                    key += line;
                }
            }
            br.close();

            return Base64.getDecoder().decode(key);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    private static byte[] rsaEncrypt(byte[] input, byte[] rsaPublicKey) {
        try {
            X509EncodedKeySpec keySpec = new X509EncodedKeySpec(rsaPublicKey);
            PublicKey pubKey = KeyFactory.getInstance("RSA").generatePublic(keySpec);

            Cipher cipher = Cipher.getInstance("RSA/ECB/PKCS1Padding");
            cipher.init(Cipher.ENCRYPT_MODE, pubKey);
            return cipher.doFinal(input);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }


}
